package com.jonah.demo.hbase.thrift;

import com.jonah.demo.hbase.thrift.common.MD5Utils;
import com.jonah.demo.hbase.thrift.config.HBaseThriftConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * # v1.0
 * create 'test_phone_md5','data','_index'
 * # v1.1 预分区
 * create 'test_phone_md5','data','_index'
 */
public class InsertDataDemo {

    private static final Logger LOGGER = LoggerFactory.getLogger(InsertDataDemo.class);

    public static void main(String[] args) throws IOException {
        // Instantiating Configuration class


        HBaseThriftConfig instant = HBaseThriftConfig.getInstant();


        // Instantiating HTable class


        String startRow = instant.getStartRow();
        Integer batchInt = instant.getBatchInt();
        String endRow = instant.getEndRow();
        Long startRowLong = instant.getStartRowLong();
        Long endRowLong = instant.getEndRowLong();
        String tableName = instant.getTableName();
        String zkHost = instant.getZkHost();
        String sig = instant.getSig();

        LOGGER.debug("startRow={}", startRow);
        LOGGER.debug("endRow={}", endRow);
        LOGGER.debug("tableName={}", tableName);
        LOGGER.debug("zkHost={}", zkHost);
        LOGGER.debug("sig={}", sig);
        LOGGER.debug("startRowLong={}", startRowLong);
        LOGGER.debug("endRowLong={}", endRowLong);
        LOGGER.debug("batchInt={}", batchInt);


        Configuration config = HBaseConfiguration.create();
        config.set(HConstants.ZOOKEEPER_QUORUM, instant.getZkHost());
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(zkHost) || StringUtils.isBlank(tableName) || startRowLong == null || endRowLong == null || batchInt == null) {
            LOGGER.error(String.format("sig=%s,zkhost=%s,tableName=%s,start=%s,startLong=%s,end=%s,endLog=%s,batchInt=%s", sig, zkHost, tableName, startRow, startRowLong, endRow, endRowLong, batchInt));
            throw new IllegalArgumentException(String.format("sig=%s,zkhost=%s,tableName=%s,start=%s,startLong=%s,end=%s,endLog=%s,batchInt=%s", sig, zkHost, tableName, startRow, startRowLong, endRow, endRowLong, batchInt));
        }


        Connection connection = ConnectionFactory.createConnection(config);
        Table hTable = connection.getTable(TableName.valueOf(tableName));
//        BufferedMutator hTable = connection.getBufferedMutator(TableName.valueOf(tableName));
        try {
            List<Put> puts = new ArrayList<>();
            for (long index = startRowLong; index < endRowLong; index++) {

                String phone = String.valueOf(index);
                String md5 = MD5Utils.getMD5Code(phone);
                LOGGER.info("start.phone={},start.md5={}", phone, md5);

//                if (!org.apache.commons.collections.CollectionUtils.isEmpty(puts) && CollectionUtils.size(puts) == batchInt) {
//                    Put start = puts.get(0);
////                    Put start_md5 = puts.get(1);
////                    Put end = puts.get(puts.size() - 2);
//                    Put end_md5 = puts.get(puts.size() - 1);
//
////                    LOGGER.info("batch ====>start={},start_md5={},end={},end_md5={}",Bytes.toString(start.getRow()),Bytes.toString(start_md5.getRow()),Bytes.toString(end.getRow()),Bytes.toString(end_md5.getRow()));
//                    LOGGER.info("batch ====>start_md5={},end_md5={}", Bytes.toString(start.getRow()), Bytes.toString(end_md5.getRow()));
//                    hTable.put(puts);
////                    List<Mutation> mus = new ArrayList<>();
////
////                    mus.addAll(puts);
////                    hTable.mutate(mus);
//
//                    puts = new ArrayList<>();
//                }


//                Put phonePut = new Put(Bytes.toBytes(phone));
//                phonePut.addColumn(Bytes.toBytes("data"), Bytes.toBytes("md5"), Bytes.toBytes(md5));
//                phonePut.addColumn(Bytes.toBytes("data"), Bytes.toBytes("phone"), Bytes.toBytes(phone));
//                phonePut.addColumn(Bytes.toBytes("_index"), Bytes.toBytes(phone), Bytes.toBytes(md5));
//                phonePut.addColumn(Bytes.toBytes("_index"), Bytes.toBytes(md5), Bytes.toBytes(phone));
//                puts.add(phonePut);

                //
//                LOGGER.debug("start.put,row={},data={}",phone,JsonUtil.toJson(phonePut));
//                hTable.put(phonePut);

//                hTable.put(phonePut);
                Put md5Put = new Put(Bytes.toBytes(md5));
                md5Put.addColumn(Bytes.toBytes("data"), Bytes.toBytes("md5"), Bytes.toBytes(md5));
                md5Put.addColumn(Bytes.toBytes("data"), Bytes.toBytes("phone"), Bytes.toBytes(phone));
                md5Put.addColumn(Bytes.toBytes("_index"), Bytes.toBytes(phone), Bytes.toBytes(md5));
                md5Put.addColumn(Bytes.toBytes("_index"), Bytes.toBytes(md5), Bytes.toBytes(phone));
//                puts.add(md5Put);
//                LOGGER.debug("start.put,row.md5={},row.phone={}",md5,phone);
//                hTable.mutateRow(puts);
//                RowMutations rm = ;
//                hTable.batch();
//                hTable.mutateRow(rm);
//                hTable.put(puts);
                hTable.put(md5Put);

//                List<Mutation> mutations = new ArrayList<Mutation>();
//                mutations.add(p);
//                table.mutate(mutations);
                //如果不flush 在后面get可能是看不见的
//                table.flush();

            }
        } catch (Exception ex) {
            ex.printStackTrace();
            LOGGER.error("全局异常", ex);
        } finally {
            hTable.close();
            connection.close();
        }

    }
}
